package net.bwie.realtime.jtp.dws.log.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class JtpTrafficPageViewMinuteWindowDwsJob3 {
    public static void main(String[] args) {
        //1.执行环境-env
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        env.enableCheckpointing(3000L);

        //2.数据源-source
        DataStream<String> pageStream = KafkaUtil.consumerKafka(env, "dwd-traffic-page-log");

        //数据转换-transformation
        handle(pageStream);

    }

    private static DataStream<String> handle(DataStream<String> stream) {
        //1.按照mid设备ID分组，用于计算UV,使用状态State记录今日是否第一次访问
        KeyedStream<String, String> midStream = stream.keyBy(
                json -> JSON.parseObject(json).getJSONObject("common").getString("mid")
        );

        //2.将流中每条日志数据封装实体类Bean对象

        //3.事件时间字段和水位线

        //4.分组keyBy:ar地区，ba品牌，ch渠道，is_new新老访客


        //5.开窗：滚动窗口，滚动窗口大小为1分钟

        //6:聚合：对窗口中数据计算

        //7.返回结果
        return null;
    }
}
